package com.lagou.spi;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;

import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

//@Activate(group = {CommonConstants.CONSUMER})
public class TPMonitorFilter implements Filter {

    /** 容量若固定则直接指定 **/
    private Map<String, ConcurrentLinkedQueue<InvokeNode>> map = new HashMap<>(3);
    /** 缓存时钟 **/
    private CurrentTimeMillisClock clock = CurrentTimeMillisClock.getInstance();
    /** 定时单线程池 **/
    private ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
    /** 当前统计时间戳 **/
    private long CURRENT_TIMESTAMP = System.currentTimeMillis();
    /** 滑动窗口1min **/
    private long SLIDING_WINDOW = 6000;

    {
        statistics();
    }

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        URL url = invoker.getUrl();
        String methodName = invocation.getMethodName();
        String serviceName = invocation.getServiceName();

        //开始计时
        long start = clock.now();  /** 并发需要使用JNI来获取精确时钟 **/

        //开始调用
        Result invoke = invoker.invoke(invocation);

        //计算耗时
        long cost = clock.now() - start;

        if (map.get(methodName) == null) {
            ConcurrentLinkedQueue queue = new ConcurrentLinkedQueue();
            queue.offer(new InvokeNode(cost, start));
            map.put(methodName, queue);
        } else {
            map.get(methodName).offer(new InvokeNode(cost, start));
        }

        return invoke;
    }

    /**
     * 每隔5s打印一次最近1分钟内每个方法的TP90、TP99的耗时情况
     *
     * 分析: 1min是一个滑动窗口大小(单向队列实现), 5s是定时任务间隔,  比如:
     *      此刻为 12:00:00 ——> 统计 11:59:00 ~ 12:00:00
     *      此刻为 12:00:05 ——> 统计 11:59:05 ~ 12:00:05
     *
     * TP90: 这一分钟内所有到达的请求, 调用耗时进行排序, 取所有请求排序后 * 0.9那一次的耗时作为 TP90
     * TP99: 同理, 取所有请求排序后 * 0.99那一次的耗时作为 TP99
     *
     * 意思是在一个时间窗口内, 比如阈值指标是500ms, 那么TP90就是这段时间内所有请求的90%都不能超过500ms, 超过就会报警
     * 同理TP99, TP999 则要求时间窗口内所有调用的99% 或 99.9% 都要低于500ms, 意味着对系统性能要求极高
     * 能做到 TP90 就是一个还算不错的系统
     *
     * 问题:
     * 1. 使用动态数组: 1min内请求数量是不固定的, 可能造成扩容; 每次检查会从头部删除造成元素移动, 若5s内请求在波峰会导致频繁移动数组元素(非常差)
     * 2. 使用 non-blocking 单向队列:  不会移动元素, 但会队列元素需要遍历得到元素并排序, 遍历是耗时操作, 1min内在波峰时可能会有非常多请求(比较差, 但比上面好一点)
     */
    void statistics() {
        executor.scheduleAtFixedRate(() -> {
            /** 更新统计时间戳 **/
            CURRENT_TIMESTAMP = System.currentTimeMillis();

            map.forEach((k, v) -> {
                if (!v.isEmpty()) {
                    List<Long> collect = v.parallelStream()
                            .filter(node -> CURRENT_TIMESTAMP - node.getTimestamp() <= SLIDING_WINDOW)
                            .map(InvokeNode::getCost)
                            .sorted(Comparator.comparing(Long::longValue))
                            .collect(Collectors.toList());
                    System.out.println(String.format("%s - TP90: %dms ,TP99: %dms", k, tp(collect, 90), tp(collect, 99)));

                    /** 移除队尾过期节点(不在滑动窗口内) **/
                    while (null != v.peek() && CURRENT_TIMESTAMP - v.peek().getTimestamp() > SLIDING_WINDOW) v.poll();
                }
            });
            System.out.println("-----------------");
        }, 0, 5000, TimeUnit.MILLISECONDS);

    }

    long tp(List<Long> list, int percent) {
        float percentF = (float)percent/100;
        int index = (int)(percentF * list.size() - 1);
        return list.get(index);
    }

    /**
     * 队列节点
     */
    static class InvokeNode {
        long cost;   /** 耗时 **/
        long timestamp; /** 调用时间 **/

        public InvokeNode(long cost, long timestamp) {
            this.cost = cost;
            this.timestamp = timestamp;
        }

        public long getCost() {
            return cost;
        }

        public void setCost(long cost) {
            this.cost = cost;
        }

        public long getTimestamp() {
            return timestamp;
        }

        public void setTimestamp(long timestamp) {
            this.timestamp = timestamp;
        }
    }

}
